package com.zc.imooc.finaltest.controller;

import com.zc.imooc.finaltest.kafka.adminClient.KafkaAdminClient;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.config.ConfigResource;
import org.springframework.web.bind.annotation.RestController;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

/**
 * @program: final-test
 * @description:
 * @author: zengchen
 * @create: 2020-09-21 09:36
 **/
@RestController
public class AdminClientTestController {

    private static final String topicName = "realNewsTopic";

    public static void main(String[] args) throws Exception {
        AdminClient adminClient = KafkaAdminClient.getAdminClient();
//        adminClient.createTopics(Collection<NewTopic> newTopics)
//        adminClient.listTopics()
//        adminClient.deleteTopics()

//        createTopic(adminClient);
//        deleteTopics(adminClient);
//        listTopics(adminClient);
//        descTopic(adminClient);

//        modifyConfig(adminClient);

//        addPartition(adminClient);

        descConfig(adminClient);

    }

    private static void addPartition(AdminClient adminClient) throws Exception {
        NewPartitions newPartitions = NewPartitions.increaseTo(1);
        Map<String, NewPartitions> newPartitionsMap = new HashMap<>();
        newPartitionsMap.put(topicName, newPartitions);
//        CreatePartitionsOptions createPartitionsOptions = new CreatePartitionsOptions();
        CreatePartitionsResult partitions = adminClient.createPartitions(newPartitionsMap);
//        System.out.println(ToStringBuilder.reflectionToString(partitions.all().get()));
    }

    private static void modifyConfig(AdminClient adminClient) throws Exception {
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
        ConfigEntry configEntry = new ConfigEntry("preallocate", "true");
        Config config = new Config(Arrays.asList(configEntry));
        Map<ConfigResource, Config> configs = new HashMap<>();
        configs.put(configResource, config);
        AlterConfigsResult alterConfigsResult = adminClient.alterConfigs(configs);
        System.out.println(ToStringBuilder.reflectionToString(alterConfigsResult.all().get()));
    }

    private static void descConfig(AdminClient adminClient) throws Exception {
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
        DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Arrays.asList(configResource));
        Config config = describeConfigsResult.values().get(configResource).get();
        System.out.println(ToStringBuilder.reflectionToString(config));
    }

    private static void descTopic(AdminClient adminClient) throws Exception {
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(topicName));
        KafkaFuture<TopicDescription> topicDescriptionKafkaFuture = describeTopicsResult.values().get(topicName);
        System.out.println(ToStringBuilder.reflectionToString(topicDescriptionKafkaFuture.get()));
    }

    private static void createTopic(AdminClient adminClient) throws Exception {
        Short replicationFactor = 1;
        NewTopic newTopic = new NewTopic(topicName, 1, replicationFactor);
        Collection<NewTopic> newTopics = Arrays.asList(newTopic);
        CreateTopicsResult topicsResult = adminClient.createTopics(newTopics);
        System.out.println(ToStringBuilder.reflectionToString(topicsResult.config(topicName).get()));
    }

    private static void listTopics(AdminClient adminClient) throws Exception {
        ListTopicsResult listTopicsResult = adminClient.listTopics();
//        ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
//        listTopicsOptions.listInternal(true); //查询包括内部的topic
//        ListTopicsResult listTopicsResult = adminClient.listTopics(listTopicsOptions);
        KafkaFuture<Collection<TopicListing>> listings = listTopicsResult.listings();
        Collection<TopicListing> topicListings = listings.get();
        if (CollectionUtils.isNotEmpty(topicListings)) {
            for (TopicListing topicListing : topicListings) {
                System.out.println(ToStringBuilder.reflectionToString(topicListing));
            }
        } else {
            System.out.println("topics is empty!");
        }
    }

    private static void deleteTopics(AdminClient adminClient) throws Exception {
        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList(topicName));
        System.out.println(ToStringBuilder.reflectionToString(deleteTopicsResult));
    }

}
